package com.xunk.equipment.config;

import com.xunk.common.core.utils.DataUtils;
import com.xunk.common.core.utils.IdUtils;
import com.xunk.common.core.utils.XunkAssert;
import com.xunk.equipment.annotation.MqttListener;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;

/**
 * @ProjectName: xunk
 * @Package: com.xunk.equipment.config
 * @ClassName: MQTTClient
 * @Author: yangwendong
 * @Description:
 * @Date: 2021/3/29 11:34
 * @Version: 1.0
 */
public class MQTTClient implements MqttCallbackExtended {

    private static final Logger log = LoggerFactory.getLogger(MQTTClient.class);

    private ScheduledExecutorService scheduler = null;
    // MQTT的连接设置
    private final MqttConnectOptions options = new MqttConnectOptions();

    private MqttClient client;
    private String clientId;
    private final String host;
    private final String userName;
    private final String passWord;


    public MQTTClient(String host, String userName, String password) {
        this.host = host;
        this.clientId = IdUtils.fastSimpleUUID();
        this.userName = userName;
        this.passWord = password;
    }


    private List<IMqttMessageListener> messageListeners;
    private ThreadPoolTaskExecutor asyncMqttTaskExecutor;
    /**
     * 监听器类型
     */
    private int type;

    public void setClientId(String clientId) {
        this.clientId = clientId;
    }

    public void setType(int type) {
        this.type = type;
    }

    public void setMessageListeners(List<IMqttMessageListener> messageListeners) {
        this.messageListeners = messageListeners;
    }

    public void setAsyncMqttTaskExecutor(ThreadPoolTaskExecutor asyncMqttTaskExecutor) {
        this.asyncMqttTaskExecutor = asyncMqttTaskExecutor;
    }

    //    /**
//     * left : topic name
//     * middle : Qos
//     * right : 处理消息
//     */
//    private static final List<Triple<String, Integer, IMqttMessageListener>> topicList = new ArrayList<>();

//    static {
//        topicList.add(Triple.of("v1/smartec/heatmeter/+", 2, new AcrelListener()));
//    }

//    public MQTTClient() throws MqttException {
//        start();
//        log.info("mqtt客户端启动成功，发布 OR 订阅 ");
//    }

    public void start() {
        // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
        try {
            client = new MqttClient(host, clientId, new MemoryPersistence());

            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(false);
            // 设置连接的用户名
            options.setUserName(userName);
            // 设置连接的密码
            options.setPassword(passWord.toCharArray());
            // 设置超时时间 单位为秒
            options.setConnectionTimeout(10);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(20);
            // 设置回调
            client.setCallback(this);
//            MqttTopic topic = client.getTopic(TOPIC);
            //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
//            options.setWill(topic, "close".getBytes(), 2, true);
            client.connect(options);

            // 订阅主题
            subscribeToTopic();
            log.info("初始化完成 {}", host);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    private void subscribeToTopic() throws MqttException {
        // Qos: 0 尽力而为，至多一次，可能丢失。1 确保到达，至少一次，可能重复。2 只有一次
//        client.subscribe(topicList.stream().map(Triple::getLeft).toArray(String[]::new),
//                topicList.stream().mapToInt(Triple::getMiddle).toArray(),
//                topicList.stream().map(Triple::getRight).toArray(IMqttMessageListener[]::new));
        XunkAssert.notEmpty(messageListeners, "需要配置监听器");
        for (IMqttMessageListener listener : messageListeners) {
            MqttListener mqttListener = listener.getClass().getAnnotation(MqttListener.class);
            if (Objects.equals(mqttListener.type(), type)) {
                client.subscribe(mqttListener.topic(), mqttListener.qos(), listener);
            }
        }
    }

    /**
     * 发布信息
     *
     * @param topic
     * @param message
     */
    public void publish(String topic, MqttMessage message) {
        CompletableFuture.runAsync(() -> {
            MqttTopic mqttTopic = client.getTopic(topic);
            try {
                MqttDeliveryToken token = mqttTopic.publish(message);
                log.info("Token is complete:" + new String(token.getMessage().getPayload()));
                token.waitForCompletion();
                log.info("Token is complete:" + token.isComplete());
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }, Objects.nonNull(asyncMqttTaskExecutor) ? asyncMqttTaskExecutor : ForkJoinPool.commonPool());
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.info("重连成功");
        closeReconnect();
    }

    @Override
    public void connectionLost(Throwable cause) {
        log.error("导致重连的原因 {}", cause.getMessage());
        startReconnect();
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        try {
            log.info("deliveryComplete---------{}", token.isComplete());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 开启重连线程
     */
    private synchronized void startReconnect() {
        log.error("mqtt重连中。。。");
//        int num = 1;
//        int time = 5000;
//        while (true) {
//            time = num * time;// 5s 10s 30s
//            if (!client.isConnected()) {
//                try {
//                    client.reconnect();
//                    // 重新订阅主题
//                    subscribeToTopic();
//                } catch (MqttException e) {
//                    e.printStackTrace();
//                }
//                num++;
//            }
//            if (client.isConnected()) {
//                break;
//            }
//            if (num >= 4) {
//                log.error("mqtt连接重试3次后依然失败");
//                break;
//            }
//            try {
//                Thread.sleep(time);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }
        if (DataUtils.isNotEmpty(scheduler)) {
            return;
        }
        scheduler = Executors.newSingleThreadScheduledExecutor();
        // 异步单线程 10s重试
        scheduler.scheduleAtFixedRate(() -> {
            if (!client.isConnected()) {
                try {
                    client.connect(options);
                    // 重新订阅主题
                    subscribeToTopic();
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }
        }, 0, 10, TimeUnit.SECONDS);

    }

    /**
     * 关闭重连线程
     */
    private synchronized void closeReconnect() {
        if (DataUtils.isNotEmpty(scheduler)) {
            scheduler.shutdownNow();
            scheduler = null;
        }
    }

}
